package com.chenjj.bigdata.flink.flinksql;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.junit.Test;

public class FlinkSqlTester {





    /**
     * 维表驱动计算结果表。
     * select a.name,b.age from user a left join age b on a.id = b.id
     *
     * user:
     * 1,zhangsan
     *
     * age:
     * 1,19
     * 1,20
     * 1,21
     *
     * 结果预期：
     * zhangsan,19
     * zhangsan,20
     * zhangsan21
     *
     * 实际结果：
     * | +I |                       zhangsan |      (NULL) |
     * | -D |                       zhangsan |      (NULL) |
     * | +I |                       zhangsan |          19 |
     * | +I |                       zhangsan |          20 |
     * | +I |                       zhangsan |          21 |
     *
     */
    @Test
    public void  test1() throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //普通 element
        //DataStreamSource<User> dsUser = env.fromElements(new User("1","zhangsan"));
        //DataStreamSource<Age> dsAge = env.fromElements(new Age("1",19),new Age("1",20),new Age("1",21));

        //socket 模拟数据的先后
        //dsUser
        DataStream<String> dsUserSocket = env.socketTextStream("192.168.152.135",9000, "\n");
        SingleOutputStreamOperator<User> dsUser = dsUserSocket.map(new MapFunction<String, User>() {
            @Override
            public User map(String value) throws Exception {
                String [] strs = value.split(",");
                return  new User(strs[0],strs[1]);
            }
        });

        //dsAge
        DataStream<String> dsAgeSocket = env.socketTextStream("192.168.152.135",9001, "\n");
        SingleOutputStreamOperator<Age> dsAge = dsAgeSocket.map(new MapFunction<String, Age>() {
            @Override
            public Age map(String value) throws Exception {
                String [] strs = value.split(",");
                return  new Age(strs[0],Integer.parseInt(strs[1]));
            }
        });

        //register table
        tableEnv.registerDataStream("t_user",dsUser,"id,name");
        tableEnv.registerDataStream("t_age",dsAge,"id,age");

        tableEnv.executeSql("select a.name,b.age  from  t_user a left join t_age b on a.id = b.id").print();

        //tableEnv.execute("test1");

    }


    /**
     * 维表驱动计算结果表，加上时间限制
     *
     *
     *
     * user:
     * 1,zhangsan,2021-11-30 12:00:00
     *
     * age:
     * 1,19,2021-11-30 12:00:01
     * 1,20,2021-12-01 12:00:02
     *
     * 结果预期：
     * zhangsan,19
     * zhangsan,20
     *
     */
    @Test
    public void test2(){
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //普通 element
//        DataStreamSource<User> dsUser = env.fromElements(new User("1","zhangsan","2021-12-01 12:00:00"));
//        DataStreamSource<Age> dsAge = env.fromElements(new Age("1",19,"2021-12-01 12:00:01"),new Age("1",20,"2021-12-02 12:00:00"));

        //socket 模拟数据的先后
        //dsUser
        DataStream<String> dsUserSocket = env.socketTextStream("192.168.152.135",9000, "\n");
        SingleOutputStreamOperator<User> dsUser = dsUserSocket.map(new MapFunction<String, User>() {
            @Override
            public User map(String value) throws Exception {
                String [] strs = value.split(",");
                return  new User(strs[0],strs[1],strs[2]);
            }
        });

        //dsAge
        DataStream<String> dsAgeSocket = env.socketTextStream("192.168.152.135",9001, "\n");
        SingleOutputStreamOperator<Age> dsAge = dsAgeSocket.map(new MapFunction<String, Age>() {
            @Override
            public Age map(String value) throws Exception {
                String [] strs = value.split(",");
                return  new Age(strs[0],Integer.parseInt(strs[1]),strs[2]);
            }
        });

        //register table
        tableEnv.registerDataStream("t_user",dsUser,"id,name,data_date");
        tableEnv.registerDataStream("t_age",dsAge,"id,age,data_date");

        tableEnv.executeSql("select a.name,b.age,a.data_date,date_format(timestampadd(hour,8,current_timestamp),'yyyy-MM-dd 00:00:00') " +
                "  from  t_user a left join t_age b on a.id = b.id" +
                "  where a.data_date > date_format(timestampadd(hour,8,current_timestamp),'yyyy-MM-dd 00:00:00')" +
                "").print();

        //tableEnv.execute("test1");
    }




    // User
    public static class User {

        public String id;
        public String name;
        public String data_date="1970-01-01 00:00:00";

        public User() {}

        public User(String id, String name) {
            this.id = id;
            this.name = name;
        }

        public User(String id, String name,String data_date) {
            this.id = id;
            this.name = name;
            this.data_date = data_date;
        }

        @Override
        public String toString() {
            return id + " ," + name;
        }
    }


    // Age
    public static class Age {

        public String id;
        public Integer age;
        public String data_date="1970-01-01 00:00:00";

        public Age() {}

        public Age(String id, Integer age) {
            this.id = id;
            this.age = age;
        }

        public Age(String id, Integer age,String data_date) {
            this.id = id;
            this.age = age;
            this.data_date = data_date;
        }

        @Override
        public String toString() {
            return id + " ," + age;
        }
    }

}
